package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
 * @author Gerry chan
 * @version 1.0
 *          2020/12/29 21:06
 * Spark Streaming关闭- 优雅的恢复数据
 * https://www.bilibili.com/video/BV11A411L7CK?p=200
 *
 * */
object SparkStreaming08_Resume {
  def main(args: Array[String]): Unit = {

    //从检查点恢复数据
    val ssc = StreamingContext.getActiveOrCreate("cp1", ()=>{
      val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
      val ssc = new StreamingContext(sparkConf, Seconds(3))

      val lines = ssc.socketTextStream("localhost", 9999)
      val wordToOne = lines.map((_,1))

      wordToOne.print()
      ssc
    })


    ssc.checkpoint("cp1")
    ssc.start()

    ssc.awaitTermination() // block阻塞main线程


  }

}
